package com.flink.sink.file;

import com.flink.entity.User;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.core.fs.Path;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;

import java.time.LocalDate;

/**
 * 描述:
 * 结果输出为文件
 *
 * @author yanzhengwu
 * @create 2022-07-22 13:38
 */
public class SinkToFile {
    public static void main(String[] args) throws Exception {

        //声明执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //测试为了保证元素顺序并行度设置为4，则每个线程生成一个文件
        env.setParallelism(4);

        DataStreamSource<User> stream = env.fromElements(
                new User("用户2", "家电", LocalDate.of(2022, 6, 2), 12, 10000L),
                new User("用户1", "零食", LocalDate.of(2022, 1, 1), 11, 10000L),
                new User("用户2", "洗发水", LocalDate.of(2022, 6, 2), 13, 10000L),
                new User("用户3", "椅子", LocalDate.of(2022, 6, 3), 14, 10000L),
                new User("用户3", "手机", LocalDate.of(2022, 6, 3), 15, 10000L),
                new User("用户1", "电脑", LocalDate.of(2022, 6, 2), 13, 10000L),
                new User("用户1", "手机", LocalDate.of(2022, 6, 16), 16, 10000L),
                new User("用户2", "洗发水", LocalDate.of(2022, 6, 2), 13, 60000L)
        );

        StreamingFileSink<String> build = StreamingFileSink.<String>forRowFormat(
                new Path("./output"), new SimpleStringEncoder<>("UTF-8"))
                //执行策略
                .withRollingPolicy(
                        //默认执行策略
                        DefaultRollingPolicy.builder()
                                //文件达到多大开始重新生成文件  这里写的是1G
                                .withMaxPartSize(1024*1024*1024)
                                //间隔多长时间开启新的文件 这里是5分钟
                                .withRolloverInterval(1000*60*5)
                                //间隔多长时间没有写入信息 则认为本文件已经结束，也就是达到了重新生成文件标准，此文件应该关闭了，这里写的是1分钟  TODO 注：这里要比上面开启新文件的时间 间隔要小
                                .withInactivityInterval(1000*60)
                                .build()
                )
                .build();
        //由于类型是 User 但是文件输出是String 所以需要map做一个转换
        stream.map(User::toString)
                .addSink(build);


        env.execute();
    }
}
